package com.sztz.batch;

import com.sztz.entity.FirmUser;
import com.sztz.entity.User;
import com.sztz.service.UserTestService;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.Order;
import org.springframework.batch.item.database.PagingQueryProvider;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.item.database.builder.JdbcPagingItemReaderBuilder;
import org.springframework.batch.item.database.support.MySqlPagingQueryProvider;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.sql.DataSource;
import java.util.HashMap;
import java.util.Map;

/**
 * SpringBatch核心配置类
 * @author yzy
 * @date 2022/8/17 11:23
 */
@Configuration
@EnableBatchProcessing
@AllArgsConstructor
@Slf4j
public class MyBatchConfiguration {
    /**
     * 注入JobBuilderFactory
     */
    private JobBuilderFactory jobBuilderFactory;
    /**
     * 注入StepBuilderFactory
     */
    private StepBuilderFactory stepBuilderFactory;
    /**
     * 数据源
     */
    private DataSource dataSource;
    /**
     * 注入用户处理服务
     */
    private UserTestService userService;
    /**
     * 注册Job执行监听器
     * @author yzy
     * @date 2022/8/17 15:12
     */
    @Bean
    public MyJobListener myJobListener(){
        return new MyJobListener();
    }
    /**
     * 注册一个用户信息批处理任务
     * @author yzy
     * @date 2022/8/17 15:12
     */
    @Bean
    public Job userJob(){
        return jobBuilderFactory.get("UserJob1").start(addUserStep()).next(sumFirmUserStep()).listener(myJobListener()).build();
    }
    /**
     * 注册批处理的一个步骤，批量添加用户信息
     * @author yzy
     * @date 2022/8/17 15:37
     */
    @Bean
    public Step addUserStep() {
        //生成一个导入文件的小步骤
        AddUserTasklet addUserTasklet = new AddUserTasklet();
        addUserTasklet.setFilePath("excel/addUser.xlsx");
        addUserTasklet.setUserService(userService);
        return stepBuilderFactory.get("addUserStep")
                .tasklet(addUserTasklet)
                .build();
    }
    /**
     * 注册批处理的一个步骤，统计每个企业的用户数量
     * @author yzy
     * @date 2022/8/17 15:37
     */
    @Bean
    public Step sumFirmUserStep() {
        return stepBuilderFactory.get("sumFirmUserStep")
                //数据量达到100时再提交到writer
                .<User,FirmUser>chunk(100)
                .reader(dbReader())
                .faultTolerant().retryLimit(3).retry(Exception.class).skip(Exception.class).skipLimit(3)
                .listener(new MyReadListener())
                .processor(userProcessor())
                .listener(new MyProcessListener())
                .writer(writer())
                .listener(new MyWriterListener())
                .build();
    }
    /**
     * 封装数据读取器
     * @author yzy
     * @date 2022/8/17 14:03
     */
    @Bean
    public ItemReader<User> dbReader() {
        //使用分页数据读取器
        return new JdbcPagingItemReaderBuilder<User>().dataSource(dataSource).name("getUserPage")
                .fetchSize(2000).pageSize(50).rowMapper(new UserRowMapper())
                .queryProvider(userProvider()).build();
    }
    /**
     * DB查询处理器
     * @author yzy
     * @date 2022/8/18 10:38
     */
    private PagingQueryProvider userProvider() {
        //多个排序字段就需要用Map
        Map<String, Order> sortKeys = new HashMap<>(8);
        sortKeys.put("USER_CODE", Order.ASCENDING);
        MySqlPagingQueryProvider provider = new MySqlPagingQueryProvider();
        //查询的字段
        provider.setSelectClause("USER_CODE, USER_NAME, SEX, PHONE, FIRM_ID ,CREATE_TIME");
        //查询的表名
        provider.setFromClause("T_USER");
        //排序
        provider.setSortKeys(sortKeys);
        return provider;
    }
    /**
     * 封装数据处理器
     * @author yzy
     * @date 2022/8/17 14:12
     */
    @Bean
    public ItemProcessor<User,FirmUser> userProcessor() {
        return (User user)->{
            FirmUser firmUser = new FirmUser();
            firmUser.setNum(1);
            return firmUser;
        };
    }
    /**
     * 封装数据输出，写入到数据库中
     * @author yzy
     * @date 2022/8/17 15:30
     */
    @Bean
    public ItemWriter<FirmUser> writer() {
        //如果使用索引参数需要 ItemPreparedStatementSetter 接口
        //命名参数且名称与User属性名称相同可以使用 BeanPropertyItemSqlParameterSourceProvider类
        return new JdbcBatchItemWriterBuilder<FirmUser>()
                .itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>())
                .sql("INSERT INTO T_FIRM_USER (FIRM_ID, NUM, SEX) VALUES(:firmId, :num, :sex) ON DUPLICATE KEY UPDATE NUM = NUM + :num")
                .dataSource(dataSource)
                .build();
    }
}
